package cn.lagou.test


import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}


object JoinDemo {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val random = scala.util.Random
    val col1 = Range(1, 50).map(idx => (random.nextInt(10), s"user$idx"))
    val col2 = Array((0, "BJ"), (1, "SH"), (2, "GZ"), (3, "SZ"), (4, "TJ"), (5, "CQ"), (6, "HZ"), (7, "NJ"), (8, "WH"), (0, "CD"))
    val rdd1: RDD[(Int, String)] = sc.makeRDD(col1)
    val rdd2: RDD[(Int, String)] = sc.makeRDD(col2)
    val rdd3: RDD[(Int, (String, String))] = rdd1.join(rdd2)
    println("-------------直接join------------------")
    println(s"rdd1.partitioner = ${rdd1.partitioner}")
    println(s"rdd2.partitioner = ${rdd2.partitioner}")
    println(s"rdd3.partitioner = ${rdd3.partitioner}")
    rdd3.glom().collect().foreach(x => println(x.toBuffer))
    println (s"rdd1.dependencies = ${rdd3.dependencies}")
    println("---------------先分区再join----------------")
    val rdd22 = rdd1.partitionBy(new HashPartitioner(3));
    val rdd33 = rdd2.partitionBy(new HashPartitioner(3));
    val rdd4: RDD[(Int, (String, String))] = rdd22.join(rdd33);
    println(s"rdd22.partitioner = ${rdd22.partitioner}")
    println(s"rdd33.partitioner = ${rdd33.partitioner}")
    println(s"rdd4.partitioner = ${rdd4.partitioner}")
    rdd4.glom().collect().foreach(x => println(x.toBuffer))
    println(s"rdd4.dependencies = ${rdd4.dependencies}")

    Thread.sleep(1000000000)
    sc.stop()
    //两个打印语句的结果是什么，对应的依赖是宽依赖还是窄依赖，为什么会是这个结果；
    //  join 操作何时是宽依赖，何时是窄依赖；

  }

}
